package com.hdu.service;


import com.alibaba.fastjson.JSON;
import com.hdu.config.DelayTaskConfig;
import com.hdu.domain.Task;
import com.hdu.enums.Priority;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static com.hdu.Constant.FUTURE_TASK_SET_KEY;
import static com.hdu.Constant.TASK_QUEUE_KEY;
import static java.lang.String.join;

@Component
public class ExecuteService {


    private static final Logger log = LoggerFactory.getLogger(ExecuteService.class);
    private final RedisService redisService;
    private final TaskService taskService;
    private final ScheduledExecutorService MOVE_FUTURE_SET_TO_TASK_QUEUE = Executors.newScheduledThreadPool(3);
    private final DelayTaskConfig delayTaskConfig;

    public ExecuteService(RedisService redisService, TaskService taskService, DelayTaskConfig delayTaskConfig) {
        this.redisService = redisService;
        this.taskService = taskService;
        this.delayTaskConfig = delayTaskConfig;
    }

    @PostConstruct
    public void init() {
        // 1.清空任务队列 + future集合, 然后同步未来 一定时间的任务到  future集合
        flushDB2FutureTaskSet(true);
        // 2.定期将 future集合里面的任务移动到 taskQueue
        moveFuture2TaskQueue();
        // 3.启动消费
        startConsumeQueue();
    }


    private void moveFuture2TaskQueue() {
        MOVE_FUTURE_SET_TO_TASK_QUEUE.scheduleWithFixedDelay(
                () -> {
                    flushDB2FutureTaskSet(false);
                    // 现在时间
                    long now = System.currentTimeMillis();
                    for (Integer priority : Priority.allSupportPriority()) {
                        String futureTaskSetKeyPattern = join(":", FUTURE_TASK_SET_KEY, priority.toString()) + "*";
                        List<String> futureTaskSetKeys = redisService.selectKeysByPattern(futureTaskSetKeyPattern);
                        for (String key : futureTaskSetKeys) {
                            List<Task> unLockTasks = redisService.zrangeByScore(Task.class, key, 0, now);
                            // 删除
                            unLockTasks.forEach(t -> {
                                // 删除
                                redisService.zrem(key, JSON.toJSONString(t));
                                // 添加到 taskQueue
                                String taskQueueKey = join(":", TASK_QUEUE_KEY, priority.toString(), t.getTaskName());
                                redisService.lpush(taskQueueKey, JSON.toJSONString(t));
                            });
                        }
                    }
                },
                10,
                10,
                TimeUnit.SECONDS
        );
    }

    private void flushDB2FutureTaskSet(boolean delete) {

        if (delete) {
            // 1. 删除以前的 futureSet 和 taskQueue
            for (Integer priority : Priority.allSupportPriority()) {
                String queuePatternKey = join(":", TASK_QUEUE_KEY, priority.toString()) + "*";
                List<String> taskQueueKeys = redisService.selectKeysByPattern(queuePatternKey);
                for (String taskQueueKey : taskQueueKeys) {
                    redisService.delete(taskQueueKey);
                }
                String futurePatternKey = join(":", FUTURE_TASK_SET_KEY, priority.toString()) + "*";
                List<String> futureSetKeys = redisService.selectKeysByPattern(futurePatternKey);
                for (String futureSetKey : futureSetKeys) {
                    redisService.delete(futureSetKey);
                }
            }
        }

        long now = System.currentTimeMillis();
        long executeTime = now + delayTaskConfig.getSupplementTaskMinute() * 60L * 1000;
        // 查出未来一段时间内会执行的任务
        List<Task> tasks = taskService.getTasksByTimeCondition(now, executeTime);
        if (Objects.nonNull(tasks)) {
            for (Task task : tasks) {
                String futureSetKey = join(":",
                        FUTURE_TASK_SET_KEY,
                        String.valueOf(task.getPriority()),
                        task.getTaskName());
                double score = Double.parseDouble(String.valueOf(task.getExecuteTime()));
                String member = JSON.toJSONString(task);
                redisService.zadd(futureSetKey, score, member);
            }
        }
    }

    private void startConsumeQueue() {
        new Thread(() -> {
            while (true) {
                log.info("try to consume");
                // 是否消费过
                boolean consumeFlag = false;
                // 遍历所有优先级
                for (Integer priority : Priority.allSupportPriority()) {
                    // taskQueue:1*
                    String patternKey = join(":", TASK_QUEUE_KEY, priority.toString()) + "*";
                    List<String> taskQueueKeys = redisService.selectKeysByPattern(patternKey);
                    for (String taskQueueKey : taskQueueKeys) {
                        Task task = redisService.rpop(taskQueueKey, Task.class);
                        if (Objects.nonNull(task)) {
                            log.info("开始消费 : task : {}", task);
                            taskService.trigger(task);
                            consumeFlag = true;
                        }
                    }
                }

                if (!consumeFlag) {
                    // 没有任何任务 休息五秒
                    try {
                        log.info("no any task, sleep 5s");
                        TimeUnit.SECONDS.sleep(5);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        }).start();
    }

}
